package com.ccc.influxdb.util;

import cn.hutool.core.util.StrUtil;
import com.ccc.influxdb.config.InfluxDBProperties;
import com.ccc.influxdb.pojo.QueryParam;
import com.ccc.influxdb.pojo.TagParam;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApi;
import com.influxdb.client.WriteOptions;
import com.influxdb.client.domain.DeletePredicateRequest;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxTable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * @author superC
 * @version 2.0.0
 * @date 2023/2/23
 */

@Component
@Slf4j
public class InfluxDBUtil {

    @Autowired
    InfluxDBClient influxDBClient;

    @Autowired
    private InfluxDBProperties properties;

    /**
     * 单条数据写入（同步）
     */
    public void write(String measurement, Map<String, Object> fields, String timestamp) {
        write(measurement, null, fields, timestamp);
    }

    public void write(String measurement, Map<String, String> tags, Map<String, Object> fields, String timestamp) {
        write(properties.getOrg(), properties.getBucket(), measurement, tags, fields, timestamp);
    }

    public void write(String orgName,
                      String bucketName,
                      String measurement,
                      Map<String, String> tags,
                      Map<String, Object> fields,
                      String timestamp) {
        long time;
        if (StrUtil.isEmpty(timestamp)) {
            time = Instant.now().toEpochMilli();
        } else {
            time = Instant.parse(timestamp).toEpochMilli();
        }
        log.info("time: {}", time);
        write(orgName, bucketName, measurement, tags, fields, time);
    }

    public void write(String orgName,
                      String bucketName,
                      String measurement,
                      Map<String, String> tags,
                      Map<String, Object> fields,
                      Long time) {
        if (time == null) {
            time = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8));
        }
        Point point = Point.measurement(measurement);
        if (tags != null && !tags.isEmpty()) {
            point.addTags(tags);
        }
        point.addFields(fields).time(time, WritePrecision.MS);
        influxDBClient.getWriteApiBlocking().writePoint(bucketName, orgName, point);
        log.info("数据写入完成");
        influxDBClient.close();
    }

    /**
     * 单条对象数据写入（同步）
     */
    public <T> void writeObj(T data) {
        writeObj(properties.getOrg(), properties.getBucket(), data);
    }

    public <T> void writeObj(String orgName, String bucketName, T data) {
        influxDBClient.getWriteApiBlocking().writeMeasurement(bucketName, orgName, WritePrecision.MS, data);
        log.info("数据写入完成");
        influxDBClient.close();
    }

    /**
     * 单条对象数据写入（异步）
     */
    public <T> void asynWriteObj(T data) {
        asynWriteObj(properties.getOrg(), properties.getBucket(), data);
    }

    public <T> void asynWriteObj(String orgName, String bucketName, T data) {
        influxDBClient.makeWriteApi().writeMeasurement(bucketName, orgName, WritePrecision.MS, data);
        log.info("数据写入完成");
        influxDBClient.close();
    }

    /**
     * 批量数据写入（同步）
     */
    public void batchWrites(List<Point> list) {
        batchWrites(properties.getOrg(), properties.getBucket(), list);
    }

    public void batchWrites(String orgName, String bucketName, List<Point> list) {
        influxDBClient.getWriteApiBlocking().writePoints(bucketName, orgName, list);
        log.info("批量数据写入完成");
        influxDBClient.close();
    }

    /**
     * 批量写入对象数据（同步）
     */
    public <T> void batchWritesObj(List<T> list) {
        batchWritesObj(properties.getOrg(), properties.getBucket(), list);
    }

    public <T> void batchWritesObj(String orgName, String bucketName, List<T> list) {
        influxDBClient.getWriteApiBlocking().writeMeasurements(bucketName, orgName, WritePrecision.MS, list);
        log.info("数据写入完成");
        influxDBClient.close();
    }

    /**
     * 批量数据写入（异步）
     */
    public void asynBatchWrites(List<Point> list) {
        asynBatchWrites(properties.getOrg(), properties.getBucket(), list);
    }

    public void asynBatchWrites(String orgName, String bucketName, List<Point> list) {
        influxDBClient.makeWriteApi().writePoints(bucketName, orgName, list);
        log.info("批量数据写入完成");
        influxDBClient.close();
    }

    /**
     * 批量写入对象数据（异步）
     */
    public <T> void asynBatchWritesObj(List<T> list) {
        asynBatchWritesObj(properties.getOrg(), properties.getBucket(), list);
    }

    public <T> void asynBatchWritesObj(String orgName, String bucketName, List<T> list) {
        influxDBClient.makeWriteApi().writeMeasurements(bucketName, orgName, WritePrecision.MS, list);
        log.info("数据写入完成");
        influxDBClient.close();
    }


    public void save(String measurement, Map<String, Object> fields) {
        WriteOptions writeOptions = WriteOptions.builder()
                .batchSize(5000)
                .flushInterval(1000)
                .bufferLimit(10000)
                .jitterInterval(1000)
                .retryInterval(5000)
                .build();
        try (WriteApi writeApi = influxDBClient.makeWriteApi(writeOptions)) {
            Point point = Point
                    .measurement(measurement)
                    .addFields(fields)
                    .time(Instant.now(), WritePrecision.MS);
            writeApi.writePoint(properties.getBucket(), properties.getOrg(), point);
        }
        log.info("批量数据写入完成");
        influxDBClient.close();
    }

    /**
     * 删除数据
     *
     * @param startTime UTC时间
     * @param stopTime  UTC时间
     * @param predicate sql where like delete statement.
     * @date 2023/3/3 14:01
     * @author superC
     */
    public void delete(String orgName, String bucketName, String startTime, String stopTime, String predicate) {
        OffsetDateTime start = OffsetDateTime.parse(startTime);
        OffsetDateTime stop = OffsetDateTime.parse(stopTime);
        influxDBClient.getDeleteApi().delete(start, stop, predicate, bucketName, orgName);
        log.info("删除数据成功");
        influxDBClient.close();
    }

    public void delete(String startTime, String stopTime, String predicate) {
        delete(properties.getBucket(), properties.getOrg(), startTime, stopTime, predicate);
    }

    public void delete(String orgName, String bucketName, OffsetDateTime start, OffsetDateTime stop, String predicate) {
        influxDBClient.getDeleteApi().delete(start, stop, predicate, bucketName, orgName);
        log.info("删除数据成功");
        influxDBClient.close();
    }

    public void delete(String orgName, String bucketName, OffsetDateTime start, OffsetDateTime stop) {
        DeletePredicateRequest deletePredicateRequest = new DeletePredicateRequest();
        deletePredicateRequest.start(start);
        deletePredicateRequest.stop(stop);

        influxDBClient.getDeleteApi().delete(deletePredicateRequest, bucketName, orgName);
        log.info("删除数据成功");
        influxDBClient.close();
    }

    public void delete(OffsetDateTime start, OffsetDateTime stop) {
        delete(properties.getBucket(), properties.getOrg(), start, stop);
    }

    public void deleteAllData(String orgName, String bucketName) {
        OffsetDateTime start = LocalDateTime.parse(FluxExpressionUtil.DEFAULT_START_TIME).atOffset(ZoneOffset.ofHours(8));
        OffsetDateTime stop = LocalDateTime.now().atOffset(ZoneOffset.ofHours(8));
        delete(orgName, bucketName, start, stop);
    }

    public void deleteMeasurementData(String orgName, String bucketName, String predicate) {
        OffsetDateTime start = OffsetDateTime.parse(FluxExpressionUtil.DEFAULT_START_TIME);
        OffsetDateTime stop = LocalDateTime.now().atOffset(ZoneOffset.ofHours(8));
        delete(orgName, bucketName, start, stop, predicate);
    }

    public void deleteMeasurementAllData(String orgName, String bucketName, String measurementName) {
        String predicate = "_measurement=\"" + measurementName + "\"";
        deleteMeasurementData(orgName, bucketName, predicate);
    }


    /**
     * 查询数据
     */
    public List<FluxTable> select(String org, String bucket, String measurement, TagParam tagParam, String startTime, String stopTime) {
        String fluxStr = queryFlux(bucket, measurement, tagParam, startTime, stopTime);
        if (StrUtil.isEmpty(org)) {
            org = properties.getOrg();
        }
        // 通过时间分组  查询时间段的数据
        return influxDBClient.getQueryApi().query(fluxStr, org);
    }

    /**
     * 查询数据
     */
    public <T> List<T> select(String org, String bucket, String measurement, TagParam tagParam, String startTime, String stopTime, Class<T> clazz) {
        String fluxStr = queryFlux(bucket, measurement, tagParam, startTime, stopTime);
        // 通过时间分组  查询时间段的数据
        return influxDBClient.getQueryApi().query(fluxStr, org, clazz);
    }

    /**
     * 查询数据
     */
    public <T> List<T> select(String org, String bucket, String[] measurements, TagParam tagParam, String startTime, String stopTime, Class<T> clazz) {
        String fluxStr = queryFlux(bucket, measurements, tagParam, startTime, stopTime);
        // 通过时间分组  查询时间段的数据
        return influxDBClient.getQueryApi().query(fluxStr, org, clazz);
    }

    /**
     * 查询数据
     */
    public <T> List<T> select(QueryParam param, Class<T> clazz) {
        String org = param.getOrg();
        if (StrUtil.isEmpty(org)) {
            org = properties.getOrg();
        }
        return influxDBClient.getQueryApi().query(queryFlux(param), org, clazz);
    }

    private String queryFlux(QueryParam param) {
        StringBuffer stringBuilder = new StringBuffer();
        if (StrUtil.isEmpty(param.getBucket())) {
            param.setBucket(properties.getBucket());
        }
        FluxExpressionUtil.appendCommonFlux(stringBuilder, param.getBucket(), param.getMeasurement(), param.getStartTime(), param.getStopTime());
        if (param.getTags() != null) {
            FluxExpressionUtil.appendTagFlux(stringBuilder, param.getTags());
        }
        if (StrUtil.isNotEmpty(param.getField())) {
            FluxExpressionUtil.appendFieldFlux(stringBuilder, param.getField());
        }
        if (param.getSortParam() != null && param.getSortParam().isSort()) {
            FluxExpressionUtil.appendSortFlux(stringBuilder, param.getSortParam().isDescFlag(), param.getSortParam().getSortField());
        }
        if (param.isGetLastData()) {
            FluxExpressionUtil.appendLastFlux(stringBuilder);
        }
        if (param.isGetTopData()) {
            FluxExpressionUtil.appendTopFlux(stringBuilder, Optional.ofNullable(param.getTopNum()).orElse(1));
        }
        if (param.isGetBottomData()) {
            FluxExpressionUtil.appendBottomFlux(stringBuilder, Optional.ofNullable(param.getBottomNum()).orElse(1));
        }
        if (param.isPivot()) {
            FluxExpressionUtil.pivot(stringBuilder);
        }

//        FluxExpressionUtil.appendTimeShiftFlux(stringBuilder);
        log.info("查询sql :{}", stringBuilder.toString());
        return stringBuilder.toString();
    }

    private String queryFlux(String bucket, String measurement, TagParam tagParam, String startTime, String stopTime) {
        return queryFlux(bucket, new String[]{measurement}, tagParam, startTime, stopTime);
    }

    private String queryFlux(String bucket, String[] measurements, TagParam tagParam, String startTime, String stopTime) {
        StringBuffer stringBuilder = new StringBuffer();
        if (StrUtil.isEmpty(bucket)) {
            bucket = properties.getBucket();
        }
        FluxExpressionUtil.appendCommonFlux(stringBuilder, bucket, measurements, startTime, stopTime);
        if (tagParam != null) {
            FluxExpressionUtil.appendTagFlux(stringBuilder, tagParam);
        }
//        FluxExpressionUtil.appendTimeShiftFlux(stringBuilder);
        log.info("查询sql :{}", stringBuilder.toString());
        return stringBuilder.toString();
    }

}